package com.allen.flink.dbus.sink;

import com.alibaba.otter.canal.protocol.FlatMessage;
import com.allen.flink.dbus.config.GlobalConfig;
import com.allen.flink.dbus.model.Flow;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;

/**
 * 功能: 写入Hbase
 *
 * @date: 2020-03-26 16:48
 * @author: Allen
 * @version: 0.0.4-snapshot
 * @Email: allenZyhang@163.com
 * @since: JDK 1.8
 **/
@Slf4j
public class HBaseSyncSink  extends RichSinkFunction<Tuple2<FlatMessage, Flow>> {
    private HBaseSyncService hbaseSyncService;
    
    /**
     * 相当于长连接
     * @param parameters
     * @throws Exception
     */
    public void open(final Configuration parameters) throws Exception {
        org.apache.hadoop.conf.Configuration hbaseConfig = HBaseConfiguration.create();
        hbaseConfig.set("hbase.zookeeper.quorum", GlobalConfig.HBASE_ZOOKEEPER_QUORUM);
        hbaseConfig.set("hbase.zookeeper.property.clientPort", GlobalConfig.HBASE_ZOOKEEPER_PROPERTY_CLIENT_PORT);
        hbaseConfig.set("zookeeper.znode.parent", GlobalConfig.ZOOKEEPER_ZNODE_PARENT);
      
        HBaseTemplate hbaseTemplate = new HBaseTemplate(hbaseConfig);
        hbaseSyncService = new HBaseSyncService(hbaseTemplate);
    }
    
    @Override
    public void close() throws Exception {
        super.close();
    }
    
    @Override
    public void invoke(final Tuple2<FlatMessage, Flow> value, final Context context) throws Exception {
        hbaseSyncService.sync(value.f1, value.f0);
    }
}
